# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
# from jms.dm.dm_sqs_false_sign_sum_dt import jms_dm__dm_sqs_false_sign_sum_dt

# doris表名
doris_table = "dm_sqs_false_sign_sum_dt"
# label名
label = f"{doris_table}_{int(time())}"
# hive中表名
hive_table = "dm_sqs_false_sign_sum_dt"
# hdfs文件路径
hdfs_path = "hdfs://{{var.value.hadoop_namespace}}/dw/hive/jms_dm.db/external/"
timeout = timedelta(hours=2).seconds
# 判断当天分区是否存在
file_path = f'"{hdfs_path}{hive_table}/dt={{{{ execution_date | cst_ds }}}}/*"'
partition_lists = []
path_lists = []
for i in range(0, 2):
    if i == 0:
        partition_lists.append("p{{ execution_date | cst_ds_nodash }}")
        path_lists.append(file_path)
    else:
        partition_lists.append("p{{ execution_date | date_add(-" + str(i) + ") | cst_ds_nodash }}")
        path_lists.append(f'"{hdfs_path}{hive_table}/dt={{{{ execution_date | date_add(-{i}) | cst_ds }}}}/*"')

partition_str = ",".join(partition_lists)
path_str = ",".join(path_lists[::-1])
doris_jms_dm__dm_sqs_false_sign_sum_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_sqs_false_sign_sum_dt',
    pool='broker_load_pool',
    email=['jokic.wang@jtexpress.com'],
    task_concurrency=1,
    conn_id='new_doris',
    pre_sql=f"""
                 TRUNCATE TABLE jms_dm.{doris_table} PARTITION ({partition_str});
                 LOAD LABEL jms_dm.{label} (
                     DATA INFILE({path_str})
                     INTO TABLE {doris_table}
                     FORMAT AS 'PARQUET' (
                                            statistic_date    ,
                                            data_type         ,
                                            agent_code        ,
                                            network_code      ,
                                            delivery_code     ,
                                            network_name      ,
                                            network_type      ,
                                            delivery_name     ,
                                            agent_name        ,
                                            virt_code         ,
                                            virt_name         ,
                                            area_info_id      ,
                                            area_info_code      ,
                                            area_info_desc    ,
                                            provider_id       ,
                                            provider_desc     ,
                                            city_id           ,
                                            city_desc         ,
                                            area_id           ,
                                            area_desc         ,
                                            is_enable         ,
                                            is_suspend        ,
                                            false_sign_cnt    ,
                                            effectiveness_cnt ,
                                            reback_type_cnt   ,
                                            third_party_cnt   ,
                                            sign_owner_cnt    ,
                                            sign_other_cnt
                                            )
                                            columns from path as (dt)
                                            set (
                                            statistic_date    =     statistic_date    ,
                                            data_type         =     data_type         ,
                                            agent_code        =     agent_code        ,
                                            network_code      =     network_code      ,
                                            delivery_code     =     delivery_code     ,
                                            network_name      =     network_name      ,
                                            network_type      =     network_type      ,
                                            delivery_name     =     delivery_name     ,
                                            agent_name        =     agent_name        ,
                                            virt_code         =     virt_code         ,
                                            virt_name         =     virt_name         ,
                                            area_info_id      =     area_info_id      ,
                                            area_info_code    =     area_info_code      ,
                                            area_info_desc    =     area_info_desc    ,
                                            provider_id       =     provider_id       ,
                                            provider_desc     =     provider_desc     ,
                                            city_id           =     city_id           ,
                                            city_desc         =     city_desc         ,
                                            area_id           =     area_id           ,
                                            area_desc         =     area_desc         ,
                                            is_enable         =     is_enable         ,
                                            is_suspend        =     is_suspend        ,
                                            false_sign_cnt    =     false_sign_cnt    ,
                                            effectiveness_cnt =     effectiveness_cnt ,
                                            reback_type_cnt   =     reback_type_cnt   ,
                                            third_party_cnt   =     third_party_cnt   ,
                                            sign_owner_cnt    =     sign_owner_cnt    ,
                                            sign_other_cnt    =     sign_other_cnt
                                            )
                 )
                 WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
                 PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

# doris_jms_dm__dm_sqs_false_sign_sum_dt << [jms_dm__dm_sqs_false_sign_sum_dt]


